package com.gzz.orchid.main.facade.controller;

import com.alibaba.fastjson2.JSONObject;
import jakarta.servlet.http.HttpServletRequest;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.UUID;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeoutException;

/**
 * @author higmd
 */
@Slf4j
@RestController
@RequestMapping("/web-flux")
@RequiredArgsConstructor
public class WebfluxDemoController {
    private final WebClient.Builder webClientBuilder;
    private final ReactiveRedisTemplate<String, String> reactiveTemplate;

    //  异步获取数据
    public Mono<String> asyncGet(String key) {
        return reactiveTemplate
                .opsForValue()
                .get(key);
    }

    //异步获取响应式流集成
    public Flux<String> reactiveKeys() {
        return reactiveTemplate
                .keys("pattern:*")
                .flatMap(key -> reactiveTemplate.opsForValue().get(key));
    }

    /**
     *
     */
    @RequestMapping("/mono")
    public Mono<String> getMono(HttpServletRequest request) {
        return Mono.just("Greetings from Spring Boot! " + LocalDateTime.now());
    }

    /**
     *
     */
    @RequestMapping("/flux")
    public Flux<String> getFlux(HttpServletRequest request) {
        Flux.just("A", "B", "C")
                .map(String::toLowerCase)
                .subscribe(System.out::println);
        return Flux.just("Greetings", " from ", "Spring Boot!", "Hi", "Aloha")
                .map(String::toUpperCase);
    }

    @GetMapping(value = "/log", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public ResponseBodyEmitter streamLogs() {
        // 设置超时时间为 60 秒
        ResponseBodyEmitter emitter = new ResponseBodyEmitter(60000L);
        //
        Executors.newSingleThreadScheduledExecutor().execute(() -> {
            try {
                // 模拟日志生成，每秒生成一条日志
                while (true) {
                    String logMessage = "Log message: [ " + LocalDateTime.now() + " ]";
                    emitter.send(logMessage);
                    // 模拟延迟，确保日志生成速度
                    Thread.sleep(1000);
                }
            } catch (InterruptedException e) {
                // 线程被中断时，关闭 Emitter
                emitter.complete();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        });
        // 设置超时处理逻辑
        emitter.onTimeout(() -> {
            log.warn("日志流连接超时。");
            emitter.completeWithError(new TimeoutException("连接超时"));
        });
        // 设置完成处理逻辑
        emitter.onCompletion(() -> {
            log.info("日志流已关闭。");
        });
        return emitter;
    }

    /**
     * &#064;CrossOrigin  注解用于解决跨域资源共享（CORS）问题。
     * 跨域请求是指浏览器从一个域名的网页去请求另一个域名的资源时，由于浏览器的同源策略会受到限制。
     * 通过该注解可以指定允许跨域访问的相关信息。
     * <p>
     * origins 属性指定允许访问该资源的源（域名），这里只允许 <a href="https://client.example.com">...</a> 这个域名的请求进行跨域访问。
     * methods 属性指定允许的 HTTP 请求方法，这里只允许 GET 和 POST 请求跨域。
     * allowedHeaders 属性指定允许的请求头，* 表示允许所有请求头。
     */
    @CrossOrigin(origins = "https://client.example.com",
            methods = {RequestMethod.GET, RequestMethod.POST},
            allowedHeaders = "*")
    @RequestMapping("/sse-event")
    public Flux<ServerSentEvent<JSONObject>> getSSE(HttpServletRequest request) {
        // 使用 WebClient 发送 SSE 请求
        WebClient webClient = webClientBuilder
                // 设置目标 API 地址
                .baseUrl("url")
                // 认证信息
                .defaultHeader(HttpHeaders.AUTHORIZATION, "Bearer " + "1233133")
                // 期望接收 SSE 格式的响应
                .defaultHeader(HttpHeaders.ACCEPT, MediaType.TEXT_EVENT_STREAM_VALUE)
                .build();
        // 用于累积 SSE 返回的消息内容
        StringBuilder contentBuilder = new StringBuilder();
        return webClient.post()
                // 发送请求体
                .bodyValue(request)
                .retrieve()
                // 解析响应数据为流式字符串
                .bodyToFlux(String.class)
                // 逐个处理 SSE 数据
                .flatMap(data -> processResponse(data, contentBuilder, request))
                // 监听前端主动断开连接
                .doOnCancel(() -> onClientCancel(contentBuilder))
                // 监听 SSE 连接终止（可能是正常结束或异常）
                .doOnTerminate(this::onStreamTerminate)
                // 发生异常时进行错误处理
                .onErrorResume(this::handleError);
    }

    /**
     * 处理 SSE 响应数据
     *
     * @param data           SSE 返回的单条数据
     * @param contentBuilder 用于累积完整的返回内容
     * @return 处理后的 SSE 事件
     */
    private Mono<ServerSentEvent<JSONObject>> processResponse(String data, StringBuilder contentBuilder, HttpServletRequest request) {
        // 判断是否为 SSE 结束标志 "[DONE]"
        if ("[DONE]".equals(data)) {
            String finalContent = contentBuilder.toString();
            log.info("SSE 消息接收完成，最终内容: {}", finalContent);
            // SSE 完成后，将内容保存到数据库
            saveToDatabase(finalContent);
            // 返回一个 SSE 事件，表示会话已完成
            return Mono.just(ServerSentEvent.<JSONObject>builder()
                    .event("done")
                    .id(UUID.randomUUID().toString())
                    // 发送空数据，仅通知前端结束
                    .data(new JSONObject())
                    .build());
        }

        //        try {
        //            // 解析 SSE 返回的 JSON 数据
        //            ChatCompletionResponse response = new ObjectMapper().readValue(data, ChatCompletionResponse.class);
        //            String content = "";
        //            if ( request.getStream() ) {
        //                content=  response.getChoices().get(0).getDelta().getContent();
        //            }else {
        //                content = response.getChoices().get(0).getMessage().getContent();
        //            }
        //            // 累积返回的消息内容
        //            if ( content != null ) {
        //                contentBuilder.append(content);
        //            }
        //
        //            // 构造 SSE 事件并返回
        //            return Mono.just(ServerSentEvent.<JSONObject>builder()
        //                    .event("add") // 事件类型
        //                    .id(UUID.randomUUID().toString()) // 生成唯一 ID
        //                    .data(new ObjectMapper().convertValue(response, JSONObject.class)) // 发送解析后的 JSON 数据
        //                    .build());
        //        } catch (JsonProcessingException e) {
        //            log.error("JSON 解析失败: {}", data, e);
        //            return Mono.just(createErrorEvent("解析失败")); // 解析异常时返回错误事件
        //        }
        return null;
    }

    /**
     * 将完整的聊天内容保存到数据库
     *
     * @param content 完整的聊天记录
     */
    private void saveToDatabase(String content) {
        if (content.isBlank()) {
            return; // 空内容不保存
        }
        log.info("消息已成功入库:{}", content);
    }

    /**
     * 处理 SSE 请求中的异常情况
     *
     * @param e 异常对象
     * @return 错误事件的 Mono
     */
    private Mono<ServerSentEvent<JSONObject>> handleError(Throwable e) {
        log.error("SSE 请求处理异常", e);
        return Mono.just(createErrorEvent("服务异常，请联系管理员"));
    }

    /**
     * 创建 SSE 错误事件
     *
     * @param message 错误信息
     * @return SSE 错误事件
     */
    private ServerSentEvent<JSONObject> createErrorEvent(String message) {
        return null;
        //        return ServerSentEvent<JSONObject>.builder()
        //                .event("error") // 事件类型为错误
        //                .id(UUID.randomUUID().toString()) // 生成唯一 ID
        //                .data(ModelMessageUtils.convertModelChatResponse(UUID.randomUUID().toString(), message)) // 发送错误信息
        //                .build();
    }

    /**
     * 监听前端主动关闭 SSE 连接
     */
    private void onClientCancel(StringBuilder builder) {
        System.out.println("11111----->" + builder);
        log.info("前端关闭了 SSE 连接，前端主动关闭 SSE 连接时触发，打印已接收内容。");
    }

    /**
     * 监听 SSE 连接终止（包括正常结束或异常）
     */
    private void onStreamTerminate() {
        log.info("SSE 连接终止，SSE 连接终止时触发，记录日志。");
    }

}
